package com.example.inventory.messaging.consumer;

import com.example.inventory.messaging.message.OrderCreatedMessage;
import com.example.inventory.messaging.message.OrderStockReservedMessage;
import com.example.inventory.messaging.producer.InventoryEventProducer;
import com.example.inventory.service.InventoryService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.stream.Collectors;

/**
 * 订单创建事件消费者
 * 监听 order_created Topic
 * 
 * @author mall-team
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class OrderCreatedConsumer {

    private final InventoryService inventoryService;
    private final InventoryEventProducer inventoryEventProducer;

    /**
     * 消费订单创建事件
     * 1. 预留库存
     * 2. 发送库存预留成功事件
     * 
     * @param message 订单创建消息
     * @param ack 手动确认对象
     */
    @KafkaListener(topics = "order_created", groupId = "inventory-service-group")
    public void handleOrderCreated(OrderCreatedMessage message, Acknowledgment ack) {
        log.info("收到订单创建事件: orderNo={}, userId={}, totalAmount={}", 
            message.getOrderNo(), message.getUserId(), message.getTotalAmount());
        
        try {
            // 1. 预留库存
            inventoryService.reserveStock(message);
            log.info("库存预留成功: orderNo={}", message.getOrderNo());
            
            // 2. 构建库存预留成功消息
            OrderStockReservedMessage reservedMessage = OrderStockReservedMessage.builder()
                    .orderNo(message.getOrderNo())
                    .userId(message.getUserId())
                    .totalAmount(message.getTotalAmount())
                    .items(message.getItems().stream()
                            .map(item -> OrderStockReservedMessage.ReservedItemInfo.builder()
                                    .skuId(item.getSkuId())
                                    .skuName(item.getSkuName())
                                    .quantity(item.getQuantity())
                                    .price(item.getPrice())
                                    .build())
                            .collect(Collectors.toList()))
                    .build();
            
            // 3. 发送库存预留成功事件
            inventoryEventProducer.sendOrderStockReservedEvent(reservedMessage);
            
            // 4. 手动确认消息
            ack.acknowledge();
            log.info("订单创建事件处理完成: orderNo={}", message.getOrderNo());
            
        } catch (Exception e) {
            log.error("处理订单创建事件失败: orderNo={}, error={}", 
                message.getOrderNo(), e.getMessage(), e);
            // 不确认消息，等待重试
            // 实际生产环境中可能需要更复杂的重试和死信队列机制
        }
    }
}

